package com.ibm.cps.model;

import java.util.ArrayList;
import java.util.Collection;

import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.gson.Gson;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.validation.ValidatorUtil;
import com.ibm.cps.newmessage.BasicDataSourceMetadata;
import com.ibm.cps.newmessage.DataSourceMetadataFactory;
import com.ibm.cps.processors.ProcessorFactory;
import com.ibm.cps.processors.ProcessorsGroup;
import com.ibm.factories.DataSourceFactory;
import com.ibm.factories.PublishProcessorFactory;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;

public abstract class MetadataAdapter {

	protected DataSourceFactory dataSourceFactory;
	protected PublishProcessorFactory publishFactory;

	public MetadataAdapter(DataSourceFactory factory, PublishProcessorFactory publishFactory) {
		this.dataSourceFactory = factory;
		this.publishFactory = publishFactory;
	}

	public String adapt2Json(String input, String source, String tenantid) throws CPSException {
		ArrayNode objects = JSONUtility.getArrayFromString(input);
		for (int i = 0; i < objects.size(); i++) {
			AdapaterFactory.validate(source, JSONUtility.fromObjectString(objects.get(i).toString()));
		}

		String output = adapt2Json(input, tenantid);
		if (output == null) {
			return null;
		}

		ArrayNode outputs = JSONUtility.getArrayFromString(output);
		ArrayNode arrayNode = JSONUtility.newArrayNode();
		for (int i = 0; i < outputs.size(); i++) {
			if (outputs.get(i).isObject()) {
				ObjectNode object = (ObjectNode) (outputs.get(i));
				addExtraInfo(object, source, tenantid);
				arrayNode.add(object);
			}else if (outputs.get(i).isArray()) {
				ArrayNode tmp_arr = (ArrayNode) outputs.get(i);
				ArrayNode addExtract_arr = JSONUtility.newArrayNode();
				for (int j = 0; j < tmp_arr.size(); j++) {
					ObjectNode object = (ObjectNode) tmp_arr.get(j);
					addExtraInfo(object, source, tenantid);
					addExtract_arr.add(object);
				}
				arrayNode.add(addExtract_arr);
			}
		}
		return arrayNode.toString();
	}

	/**
	 * get data source for input
	 *
	 * @param array
	 * @param tenantid
	 * @throws CPSException
	 */
	private void processInputProcess(ObjectNode objectNode, String tenantid) throws CPSException {
		String dataSourceMetadata = replicateDataSource(objectNode, dataSourceFactory);
		if (dataSourceMetadata != null) {
			ObjectNode dataSource = JSONUtility.fromObjectString(dataSourceMetadata);
			objectNode.setAll(dataSource);
		}
	}

	private void processorPublishProcessor(ObjectNode object, String tenantid) throws CPSException {
		JSONUtility.checkNotNull(object, MessageFields.TYPE);
		String type = JSONUtility.GetString(MessageFields.TYPE, object);
		if (!type.equals(MessageFields.PUBLISH_TYPE)) {
			return;
		}
		JSONUtility.checkNotNull(object, MessageFields.NAME);
		String name = JSONUtility.GetString(MessageFields.NAME, object);
		ObjectNode outputProcessor = publishFactory.getOutputProcessor(name, tenantid);
		// modify by gxj to match metaString
		//		object.setAll(outputProcessor);
		object.put("metadataString",outputProcessor.toString());
		object.put(MessageFields.TYPE, MessageFields.PUBLISH_TYPE);
	}

	private void processorStaticJointProcessor(ObjectNode object, String tenantid) throws CPSException {
		JSONUtility.checkNotNull(object, MessageFields.TYPE);
		String type = JSONUtility.GetString(MessageFields.TYPE, object);
		if (!type.equals(MessageFields.STATIC_JOINT)) {
			return;
		}
		JSONUtility.checkNotNull(object, MessageFields.NAME);

		String name = JSONUtility.GetString(MessageFields.NAME, object);
		object.put(MessageFields.NAME, name);

		String dataSourceMetadata = replicateStaticDataSource(object, dataSourceFactory);
		if (dataSourceMetadata != null) {
			ObjectNode dataSource = JSONUtility.fromObjectString(dataSourceMetadata);
			object.put("metadataString", dataSource.toString());
		}
		object.put(MessageFields.TYPE, type);
	}

	public static String replicateDataSource(ObjectNode objectNode, DataSourceFactory factory) throws CPSException {
		if (factory == null) {
			throw new CPSException(ErrorCode.NULL_FIELD_ERROR, "datasource factory is null");
		}
		String type = JSONUtility.GetString(MessageFields.TYPE, objectNode);
		ValidatorUtil.checkNotNull(MessageFields.TYPE, type);
		// ugly code. need to refactor.
		if (!ProcessorFactory.isInput(type)) {
			return null;
		}
		String topicName = JSONUtility.GetString(MessageFields.NAME, objectNode);
		ValidatorUtil.checkNotNull(MessageFields.NAME, topicName);
		String dataSourceMetadata = factory.replicateDataSource(topicName);
		return dataSourceMetadata;
	}


	/*
	 * add by gxj to get static datasource metadata
	 */
	public static String replicateStaticDataSource(ObjectNode objectNode, DataSourceFactory factory) throws CPSException {
		if (factory == null) {
			throw new CPSException(ErrorCode.NULL_FIELD_ERROR, "datasource factory is null");
		}
		String type = JSONUtility.GetString(MessageFields.TYPE, objectNode);
		ValidatorUtil.checkNotNull(MessageFields.TYPE, type);
		// ugly code. need to refactor.
		if (!ProcessorFactory.isStaticData(type)) {
			return null;
		}
		String topicName = JSONUtility.GetString(MessageFields.NAME, objectNode);
		ValidatorUtil.checkNotNull(MessageFields.NAME, topicName);
		BasicDataSourceMetadata metadata = factory.getDataSourceMetadata(topicName);
		String dataSourceMetadata = metadata.getInput();
		return dataSourceMetadata;
	}

	private void addExtraInfo(ObjectNode object, String source, String tenantid) throws CPSException {
		autoAppendIdField(object);
		addField2Object(object, MessageFields.SOURCE, source);
		addField2Object(object, MessageFields.TENANTID, tenantid);
		processInputProcess(object, tenantid);
		processorPublishProcessor(object, tenantid);
		processorStaticJointProcessor(object, tenantid);
	}

	private void autoAppendIdField(ObjectNode object) throws CPSException {
		String type = JSONUtility.GetString(MessageFields.TYPE, object);
		JSONUtility.checkNotNull(object, MessageFields.TYPE);
		if (ProcessorsGroup.isGroupType(type)) {
			JSONUtility.assignRandomValue(object, MessageFields.GROUPID);
		} else if (DataSourceMetadataFactory.isDataSourceType(type)) {
			JSONUtility.assignRandomValue(object, MessageFields.NAME);
		} else if (PublishProcessorFactory.isPublishProcessor(type)) {
			JSONUtility.assignRandomValue(object, MessageFields.PROCESSORID);
		} else if (ProcessorFactory.isProcessorType(type)) {
			JSONUtility.assignRandomValue(object, MessageFields.PROCESSORID);
		} else {
			throw new CPSException(ErrorCode.INVALID_INPUT_ERROR, "Unknown metadata type " + type);
		}
	}

	private void addField2Object(ObjectNode object, String fieldName, String fieldValue) {
		object.put(fieldName, fieldValue);
	}

	abstract public String adapt2Json(String input, String tenantid) throws CPSException;

	public Collection<String> getJsons(Collection<String> allMetadatas, Collection<String> targetMetadatas)
			throws CPSException {
		Collection<String> outputs = adaptFromJson(allMetadatas, targetMetadatas);
		Collection<String> results = new ArrayList<String>();
		if (outputs == null) {
			return null;
		}
		for (String output : outputs) {
			ObjectNode node = JSONUtility.fromObjectString(output);
			node.remove(MessageFields.SOURCE);
			node.remove(MessageFields.TENANTID);
			results.add(node.toString());
		}
		return results;
	}

	abstract public Collection<String> adaptFromJson(Collection<String> allMetadatas,
			Collection<String> targetMetadatas) throws CPSException;
}
